package com.bsoft.advance.client.transport;

import com.bsoft.advance.client.common.RequestMetadata;
import com.bsoft.advance.core.codec.RpcDecoder;
import com.bsoft.advance.core.codec.RpcEncoder;
import com.bsoft.advance.core.common.RpcRequest;
import com.bsoft.advance.core.common.RpcResponse;
import com.bsoft.advance.core.protocol.MessageProtocol;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 基于netty实现的rpc传输类
 */
@Slf4j
public class NettyRpcTransport implements RpcRequestTransport{
    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;
    private final RpcResponseHandler handler;

    public NettyRpcTransport() {
        // 创建统一的业务处理线程组
        eventLoopGroup = new NioEventLoopGroup();
        // 客户端启动类
        bootstrap = new Bootstrap();
        // 响应数据处理类
        handler = new RpcResponseHandler();
        // 启动类型配置
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {
                        sc.pipeline()
                                // netty心跳处理类
                                .addLast(new IdleStateHandler(0,5,0, TimeUnit.SECONDS))
                                .addLast(new RpcDecoder())
                                .addLast(new RpcEncoder<MessageProtocol>())
                                .addLast(handler);
                    }
                });
    }

    /**
     * 请求服务端通信方法
     *
     * @param metadata
     */
    @Override
    public MessageProtocol<RpcResponse> send(RequestMetadata metadata) throws Exception{

        MessageProtocol<RpcRequest> protocol = metadata.getProtocol();
        CompletableFuture<MessageProtocol<RpcResponse>> future = new CompletableFuture<>();
        // 存入响应缓存，通过RequestId把请求和响应绑定起来
        UnsolvedResponseCache.put(protocol.getHeader().getRequestId(), future);

        // 获取通道
        Channel channel = getChannel(metadata.getSocketAddress());
        // 发送消息给服务端
        if(channel.isActive()){
            channel.writeAndFlush(protocol);
        }

        // 同步等待结果
        return metadata.getTimeout() > 0 ? future.get(metadata.getTimeout(), TimeUnit.MILLISECONDS) : future.get();
    }

    /**
     * 创建连接并获取通道
     */
    private Channel createChannel(InetSocketAddress socketAddress) throws InterruptedException {
        // 连接服务端地址和端口
        ChannelFuture channelFuture = bootstrap.connect(socketAddress).sync();
        // 注册监听器，监听通道连接结果
        channelFuture.addListener(future -> {
            if (channelFuture.isSuccess()) {
                log.info("connect success to server[{}]", socketAddress.toString());
            } else {
                log.error("connect failed to server[{}]", socketAddress.toString());
                channelFuture.cause().printStackTrace();
                eventLoopGroup.shutdownGracefully();
            }
        });
        // 连接成功后，通过future拿到具体通道
        Channel channel = channelFuture.channel();
        return channel;
    }

    /**
     * 获取通道
     *
     * @return
     * @throws InterruptedException
     */
    @Override
    public Channel getChannel(InetSocketAddress socketAddress) throws Exception {
        Channel channel = ChannelProvider.get(socketAddress);
        if (channel == null) {
            Channel newChannel = createChannel(socketAddress);
            // 存入channelMap缓存中
            ChannelProvider.put(socketAddress, newChannel);
            return newChannel;
        }
        return channel;
    }

}
